package com.mccken.rapid.core.netty.processor;

import com.lmax.disruptor.dsl.ProducerType;
import com.mccken.rapid.common.current.queue.flusher.ParallelFlusher;
import com.mccken.rapid.common.enums.ResponseCode;
import com.mccken.rapid.core.RapidConfig;
import com.mccken.rapid.core.context.HttpRequestWrapper;
import com.mccken.rapid.core.helper.ResponseHelper;

import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaderValues;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpUtil;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

/**
 * @program: rapid_gateway
 * @description:
 * @author: mccken
 * @create: 2022-01-05 14:01
 **/
@Slf4j
public class NettyBatchEventProcessor implements NettyProcessor {

	private static final String THREAD_NAME_PREFIX = "rapid-flusher-";
	
	@Getter
	private final RapidConfig rapidConfig;

	private final NettyCoreProcessor nettyCoreProcessor;
	
	private final ParallelFlusher parallelFlusher;

	public NettyBatchEventProcessor(RapidConfig rapidConfig, NettyCoreProcessor nettyCoreProcessor) {
		this.rapidConfig = rapidConfig;
		this.nettyCoreProcessor = nettyCoreProcessor;
		
		this.parallelFlusher = new ParallelFlusher.Builder<HttpRequestWrapper>()
				.setEventListener(new BatchEventProcessorListener())
				.setNamePrefix(THREAD_NAME_PREFIX)
				.setThreads(rapidConfig.getProcessThread())
				.setBufferSize(rapidConfig.getBufferSize())
				.setProducerType(ProducerType.MULTI)
				.setWaitStrategy(rapidConfig.getATureWaitStrategy())
				.build();
	}

	/**
	 * 监听事件的处理核心逻辑
	 */
	public class BatchEventProcessorListener implements ParallelFlusher.EventListener<HttpRequestWrapper> {

		@Override
		public void onEvent(HttpRequestWrapper event) throws Exception {
			nettyCoreProcessor.process(event);
		}

		@Override
		public void onException(Throwable ex, long sequence, HttpRequestWrapper event) {
			HttpRequest request = event.getFullHttpRequest();
			ChannelHandlerContext ctx = event.getCtx();
			
			try {
				log.error("#BatchEventProcessorListener# onException 请求处理失败， request:{}, errorMessage: {}", request, ex.getMessage(), ex);
				
				// 首先构建响应对象
				FullHttpResponse fullHttpResponse = ResponseHelper.getHttpResponse(ResponseCode.INTERNAL_ERROR);
				
				if (!HttpUtil.isKeepAlive(request)) {
					ctx.writeAndFlush(fullHttpResponse).addListener(ChannelFutureListener.CLOSE);
				} else {
					// 如果保持链接，则需要设置一下响应头 key: CONNECTION   value: KEEP_ALIVE
					fullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
					ctx.writeAndFlush(fullHttpResponse);
				}
				
				
			} catch (Exception e) {
				log.error("#BatchEventProcessorListener# onException 请求回写失败， request:{}, errorMessage: {}", request, e.getMessage(), e);
			}
			
		}
	}

	@Override
	public void process(HttpRequestWrapper httpRequestWrapper) {
		System.out.println("NettyBatchEventProcessor add");
		this.parallelFlusher.add(httpRequestWrapper);
	}

	@Override
	public void start() {
		this.nettyCoreProcessor.start();
		this.parallelFlusher.start();
	}

	@Override
	public void shutdown() {
		this.nettyCoreProcessor.shutdown();
		this.parallelFlusher.shutdown();
	}
}
